package com.swak.vertx.transport.vertx;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.OS;
import com.swak.reactivex.context.EndPoints;
import com.swak.reactivex.context.EndPoints.EndPoint;
import com.swak.spi.ExtensionLoader;
import com.swak.utils.Lists;
import com.swak.vertx.config.VertxConfig;
import com.swak.vertx.transport.ServerVerticle;
import com.swak.vertx.transport.codec.Msg;
import com.swak.vertx.transport.codec.MsgCodec;
import com.swak.vertx.transport.vertx.endpoint.FluxEndPoint;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Promise;

/**
 * 响应式服务的启动，对于 vertx 来说， 所有的 Verticle 都在这个服务中发布
 *
 * @author: lifeng
 * @date: 2020/3/29 21:17
 */
public class MainVerticle extends AbstractVerticle implements ServerVerticle {

	protected Logger logger = LoggerFactory.getLogger(ServerVerticle.class);

	private static ExtensionLoader<FluxEndPoint> FluxEndPointLoader = ExtensionLoader
			.getExtensionLoader(FluxEndPoint.class);

	private final VertxConfig config;

	public MainVerticle(VertxConfig config) {
		this.config = config;

		// 服务器地址
		String hostName = config.getHost();
		if (!Constants.LOCALHOST.equals(hostName)) {
			hostName = OS.ip();
		}
		EndPoints.me().setHost(hostName);
	}

	/**
	 * 异步的启动
	 */
	@Override
	@SuppressWarnings({ "rawtypes" })
	public void start(Promise<Void> startPromise) {
		List<Future> futures = Lists.newArrayList();

		// 自定义一些配置
		vertx.eventBus().registerDefaultCodec(Msg.class, new MsgCodec());

		// 加载所有的端点
		List<FluxEndPoint> endpoints = FluxEndPointLoader.loadAll(this.getClass().getClassLoader());
		for (FluxEndPoint ep : endpoints) {
			futures.addAll(ep.start(vertx, config));
		}

		// 等待所有服务全部启动完成
		CompositeFuture.all(futures).onComplete(res -> {
			if (res.succeeded()) {
				List<EndPoint> endPoints = Lists.newArrayList();
				res.result().list().forEach(endpoint -> {
					if (endpoint != null && endpoint instanceof EndPoint) {
						endPoints.add((EndPoint) endpoint);
					}
				});
				EndPoints.me().setEndPoints(endPoints);
				startPromise.complete();
			} else {
				startPromise.fail(res.cause());
			}
		});
	}
}